package com.fly.demo.web.sse;

import org.apache.commons.lang3.time.DateFormatUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import reactor.core.publisher.Flux;

/**
 * 使用Spring Boot Starter Integration整合
 * 
 */
@RestController
public class SSE4Controller
{
    @Autowired
    private PublishSubscribeChannel sseChannel;
    
    @GetMapping(path = "/sse/v4", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> streamSseMvc()
    {
        return Flux.create(sink -> {
            MessageHandler messageHandler = message -> sink.next(ServerSentEvent.builder((String)message.getPayload()).build());
            // 取消订阅
            sink.onDispose(() -> sseChannel.unsubscribe(messageHandler));
            // 订阅消息
            sseChannel.subscribe(messageHandler);
        });
    }
    
    @Scheduled(cron = "0/10 * 7-23 * * ?")
    private void publish()
    {
        String message = DateFormatUtils.format(System.currentTimeMillis(), "yyyy-MM-dd HH:mm:ss.SSS");
        sseChannel.send(new GenericMessage<>(message));
    }
}